package com.googlecode.lazyrecords.hazelcast;

import com.googlecode.lazyrecords.AbstractRecords;
import com.googlecode.lazyrecords.Definition;
import com.googlecode.lazyrecords.Record;
import com.googlecode.lazyrecords.SourceRecord;
import com.googlecode.lazyrecords.mappings.StringMappings;
import com.googlecode.lazyrecords.memory.Mapords;
import com.googlecode.totallylazy.*;
import com.googlecode.totallylazy.collections.ImmutableList;
import com.googlecode.totallylazy.collections.ImmutableSortedMap;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IMap;

import java.util.Map;
import java.util.UUID;

import static com.googlecode.totallylazy.Callables.value;
import static com.googlecode.totallylazy.Maps.pairs;
import static com.googlecode.totallylazy.collections.ImmutableList.constructors.list;
import static com.googlecode.totallylazy.collections.ImmutableSortedMap.constructors.sortedMap;

public class HazelcastRecords extends AbstractRecords {
    private final HazelcastInstance instance;
    private final Mapords mapords;

    public HazelcastRecords(HazelcastInstance instance, StringMappings mappings) {
        this.instance = instance;
        this.mapords = new Mapords(mappings);
    }

    @Override
    public Number add(Definition definition, Sequence<Record> records) {
        IMap<UUID, Map<String, String>> map = mapFor(definition);
        for (Record record : records) {
            map.put(UUID.randomUUID(), mapords.asMap(definition).apply(record));
        }
        return records.size();
    }

    @Override
    public Number remove(Definition definition, Predicate<? super Record> predicate) {
        Sequence<UUID> uuids = get(definition).filter(predicate).<SourceRecord<UUID>>unsafeCast().map(value(UUID.class));
        IMap<UUID, Map<String, String>> map = mapFor(definition);
        for (UUID uuid : uuids) map.remove(uuid);
        return uuids.size();
    }

    @Override
    public Sequence<Record> get(final Definition definition) {
        return pairs(mapFor(definition)).map(new Function1<Pair<UUID, Map<String, String>>, Record>() {
            @Override
            public Record call(Pair<UUID, Map<String, String>> pair) throws Exception {
                return SourceRecord.record(pair.first(), mapords.asRecord(definition).call(pair.second()));
            }
        });
    }

    private IMap<UUID, Map<String, String>> mapFor(Definition definition) {
        return instance.getMap(definition.name());
    }
}
